package cn.simon.free.redis.mq.core.producer;

import cn.simon.free.redis.mq.annotations.RedisMessageConsumer;
import cn.simon.free.redis.mq.config.properties.RedisMessageProperties;
import cn.simon.free.redis.mq.core.AbstractRedisMessage;
import cn.simon.free.redis.mq.exceptions.NormalException;
import cn.simon.free.redis.mq.model.Message;
import com.google.common.base.Joiner;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;

import java.util.concurrent.TimeUnit;

/**
 * @author simon
 * @date 2021/7/12 17:04
 * @copyright:
 * @modifier
 * @motto a no talent, people who have to study hard
 */
public abstract class AbstractRedisMessageProducer<T extends Message> extends AbstractRedisMessage implements Producer<T> {

    private RedisMessageProperties redisMessageProperties;

    /**
     * push消息至redis
     *
     * @param queueName
     * @param message
     * @param delayTime
     * @return
     */
    protected abstract String sendMsg(String queueName, T message, Long delayTime);

    @Override
    public String send(String queueName, T message) {
        return this.send(queueName, message, 0);
    }

    @Override
    public String send(String queueName, T message, long delayTime) {
        return this.send(queueName, message, delayTime, TimeUnit.SECONDS);
    }

    @Override
    public String send(String queueName, T message, long delayTime, TimeUnit timeUnit) {
        switch (timeUnit) {
            case DAYS:
            case HOURS:
            case MINUTES:
            case SECONDS:
                return this.sendMsg(hash(queueName, message), message, System.currentTimeMillis() + timeUnit.toMillis(delayTime));
            default:
                throw new NormalException("不支持的时间格式");
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        super.setApplicationContext(applicationContext);
        this.redisMessageProperties = applicationContext.getBean(RedisMessageProperties.class);
    }

    /**
     * hash 取模,确认槽位置
     *
     * @param queueName
     * @param message
     * @return
     */
    private String hash(String queueName, Message message) {
        RedisMessageConsumer redisMessageConsumer = redisMessageProperties.getQueueNameWithHash().get(queueName);
        int corePoolSize = redisMessageConsumer.singleThreadProcess() ? 1 : redisMessageConsumer.corePoolSize();
        return Joiner.on("-").join(queueName, Math.abs(message.hashCode() % corePoolSize));
    }

}
